package com.ky.pro.java.mq.kmq.core;

import com.ky.pro.java.mq.kmq.repository.KmqMessageRepository;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public final class KmqBroker { // Broker+Connection

    KmqMessageRepository repository;

    public static final int CAPACITY = 10000;

    private final Map<String, KQueue> kmqMap = new ConcurrentHashMap<>(64);

    public void createTopic(String name) {
        kmqMap.putIfAbsent(name, new KQueue(CAPACITY, name, repository));
    }

    public KQueue findKmq(String topic) {
        return this.kmqMap.get(topic);
    }

    public KmqProducer createProducer() {
        return new KmqProducer(this);
    }

    public KmqConsumer createConsumer() {
        return new KmqConsumer(this);
    }

    public void setRepository(KmqMessageRepository repository) {
        this.repository = repository;
    }

}
